package gu.simplemq.stomp;

import static gu.simplemq.stomp.StompPoolLazys.createInstance;
import static gu.simplemq.stomp.PropertiesHelper.SHELPER;

import java.io.IOException;
import java.net.URI;
import java.net.URISyntaxException;
import java.util.Map;
import com.google.common.net.HostAndPort;

import gu.simplemq.BaseMessageQueueFactory;
import gu.simplemq.IConsumer;
import gu.simplemq.IMessageQueueFactory;
import gu.simplemq.IProducer;
import gu.simplemq.IPublisher;
import gu.simplemq.ISubscriber;
import gu.simplemq.MQLocationType;
import gu.simplemq.MQProperties;
import gu.simplemq.MessageQueueType;
import gu.simplemq.mqtt.MqttConstants;
import gu.simplemq.utils.URISupport;

/**
 * {@link IMessageQueueFactory}接口的STOMP实现类
 * @author guyadong
 *
 */
public final class MessageQueueFactoryImpl extends BaseMessageQueueFactory<StompPoolLazy> 
	implements StompConstants,MqttConstants {

	private volatile MQProperties props;

	private IMessageQueueFactory mqttFactory;
	public MessageQueueFactoryImpl() {
	}

	@SuppressWarnings({ "unchecked", "rawtypes", "resource" })
	@Override
	protected void doInit(Map<String, Object> properties){
		mqConnParams = (Map)properties;
		props =  SHELPER.with(MQLocationType.QUEUE).initParameters(properties);
		pool = createInstance(props);

		String pubsub = SHELPER.with(MQLocationType.PUBSUB).getLocationAsString(properties);
		if(pubsub != null){
			URI uri = URI.create(pubsub);
			if(MQTT_SCHEMA.equals(uri.getScheme())){
				try {
					uri = URISupport.changeScheme(uri, PAHO_MQTT_SCHEMA);
				} catch (URISyntaxException e) {
					throw new RuntimeException(e);
				}
				MQProperties p = gu.simplemq.mqtt.PropertiesHelper.MHELPER.initParameters(properties).initURI(uri);
				mqttFactory = new gu.simplemq.mqtt.MessageQueueFactoryImpl()
						.init((Map)p);
			}
		}
	
	}
	@Override
	protected HostAndPort doGetHostAndPort() {
		return SHELPER.with(MQLocationType.QUEUE).getHostAndPort(props);
	}

	@Override
	public ISubscriber getSubscriber() {
		if(mqttFactory != null){
			return mqttFactory.getSubscriber();
		}
		return StompFactory.getSubscriber(getPool());
	}

	@Override
	public IPublisher getPublisher() {
		if(mqttFactory != null){
			return mqttFactory.getPublisher();
		}
		return StompFactory.getPublisher(getPool());
	}

	@Override
	public IProducer getProducer() {
		return StompFactory.getProducer(getPool());
	}

	@Override
	public IConsumer getConsumer() {
		return StompFactory.getConsumer(getPool());
	}

	@Override
	public final MessageQueueType getImplType(){
		return MessageQueueType.ACTIVEMQ;
	}

	@Override
	public boolean testConnect(){
		return SHELPER.testConnect(props,null);
	}
	
	@Override
	public void close() throws IOException {
		if(mqttFactory != null){
			((gu.simplemq.mqtt.MessageQueueFactoryImpl)mqttFactory).close();
		}
		super.close();
	}
}
